package edu.csl.study.spark.basic

import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object StructuredStreaming_csv {
  val rootDir = System.getProperty("user.dir")+ "\\testFile\\"
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().appName("StructuredStreaming").master("local[2]").getOrCreate()
    spark.sparkContext.setLogLevel("WARN")

    val user:DataFrame = source_CSV(spark)

    import spark.implicits._
    user.createTempView("t_user")
    val result:DataFrame = spark.sql("select sum(id)  from t_user  ")

    /**
     * 3中输出模式：
     *  Complete表示结果全量输出,这种模式必须存在聚合操作
     *  Append 追加模式  非Watermark模式下，不能使用任何聚合操作
     *  Update 更新模式 Spark 2.1.1之后支持，如果查询结果中没有聚合操作，那么等同于Append模式。
     */
    val query:StreamingQuery = result.writeStream.outputMode("complete")
      .trigger(Trigger.ProcessingTime(0))//0表示处理完数据后，立即处理下一批
      .format("console")
      .start()

    query.awaitTermination();

  }


  /**
   * 数据来源- CSV
   * @param spark
   * @return
   */
  def source_CSV(spark:SparkSession):DataFrame = {
    val schema=StructType(
        StructField("id",StringType)::
        StructField("name",StringType)::
        StructField("age",IntegerType)::
        StructField("sex",IntegerType)::
        StructField("phone",IntegerType)::Nil
    )
    val path = "hdfs://192.168.100.20:9000/practise/kaikeba/StructuredStreaming_csv"
    //必须是一个目录，监控目录下的文件变化。
    val source:DataFrame = spark.readStream.schema(schema).csv(path);
    source
  }
}
